package com.itheima.mqtt.client;

import com.itheima.mqtt.enums.QosEnum;
import com.itheima.mqtt.properties.MqttProperties;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.HashSet;
import java.util.Set;


@Component
public class EmqClient {

    private static final Logger log = LoggerFactory.getLogger(EmqClient.class);


    private IMqttClient mqttClient;

    @Autowired
    private MqttProperties mqttProperties;

    @Autowired
    private MqttCallback mqttCallback;

    @lombok.Getter
    private Set<String> subscribedTopics = new HashSet<>();

    @PostConstruct
    public void init() {
        MqttClientPersistence mempersitence = new MemoryPersistence();
        try {
            mqttClient = new MqttClient(mqttProperties.getBrokerUrl(), mqttProperties.getClientId(), mempersitence);
            connect(mqttProperties.getUsername(), mqttProperties.getPassword());
            subscribe("testtopic/#", QosEnum.QoS2);
        } catch (MqttException e) {
            log.error("初始化客户端mqttClient对象失败,errorMsg={},brokerUrl={},clientId={}", e.getMessage(), mqttProperties.getBrokerUrl(), mqttProperties.getClientId());
        }
    }

    /**
     * 连接broker
     *
     * @param username username
     * @param password password
     */
    private void connect(String username, String password) {
        log.info("开始连接服务端,brokerUrl={},clientId={}", mqttProperties.getBrokerUrl(), mqttProperties.getClientId());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setCleanSession(true);
        //设置方法回调
        mqttClient.setCallback(mqttCallback);
        try {
            mqttClient.connect(options);
        } catch (MqttException e) {
            log.error("mqtt客户端连接服务端失败,失败原因{}", e.getMessage());
        }
    }


    /**
     * 断开连接
     */
    @PreDestroy
    public void disConnect() {
        try {
            log.info("开始断开连接");
            mqttClient.disconnect();
        } catch (MqttException e) {
            log.error("断开连接产生异常,异常信息{}", e.getMessage());
        }
    }

    /**
     * 重连
     */
    public void reConnect() {
        try {
            log.info("开始重新连接");
            mqttClient.reconnect();
        } catch (MqttException e) {
            log.error("重连失败,失败原因{}", e.getMessage());
        }
    }

    /**
     * 发布消息
     *
     * @param topic  主题
     * @param msg    消息
     * @param qos    qos
     * @param retain retain
     */
    public void publish(String topic, String msg, QosEnum qos, boolean retain) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        mqttMessage.setQos(qos.value());
        mqttMessage.setRetained(retain);
        try {
            log.info("开始发布消息,topic={},msg={},qos={},retain={}", topic, msg, qos.value(), retain);
            mqttClient.publish(topic, mqttMessage);
        } catch (MqttException e) {
            log.error("发布消息失败,errorMsg={},topic={},msg={},qos={},retain={}", e.getMessage(), topic, msg, qos.value(), retain);
        }
    }

    /**
     * 订阅
     *
     * @param topicFilter 订阅的主题(过滤主题)
     * @param qos         qos
     */
    public void subscribe(String topicFilter, QosEnum qos) {
        try {
            log.info("订阅主题成功!!! topicFilter={},qos={}", topicFilter, qos.value());
            mqttClient.subscribe(topicFilter, qos.value());
            subscribedTopics.add(topicFilter);
        } catch (MqttException e) {
            log.error("订阅主题失败,errorMsg={},topicFilter={},qos={}", e.getMessage(), topicFilter, qos.value());
        }

    }

    /**
     * 取消订阅
     *
     * @param topicFilter 取消订阅的主题(过滤主题)
     */
    public void unSubscribe(String topicFilter) {
        try {
            log.info("取消订阅主题!!! topicFilter={}", topicFilter);
            mqttClient.unsubscribe(topicFilter);
            subscribedTopics.remove(topicFilter);
        } catch (MqttException e) {
            log.error("取消订阅失败,errorMsg={},topicFilter={}", e.getMessage(), topicFilter);
        }
    }

}
